package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Created by Smexy on 2023/3/1
 * 
 *  Join.
 *      先Join再用窗口运算，称为 WindowJoin!
 *
 *      开窗运算，99%都是基于 EventTime
 */
public class Demo8_WindowJoin
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        SingleOutputStreamOperator<WaterSensor> ds2 = env
            .socketTextStream("hadoop103", 8889)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        // // join的条件。 on t1.id = t2.id
        ds1.join(ds2)
           //on t1.id
           .where(WaterSensor::getId)
           //= t2.id
           .equalTo(WaterSensor::getId)
           .window(TumblingEventTimeWindows.of(Time.seconds(5)))
           .apply(new JoinFunction<WaterSensor, WaterSensor, String>()
           {
               /*
                    apply(): 等窗口关闭时，才会执行一次。
                      当两个流中的数据 join上后，会调用此方法！
                */
               @Override
               public String join(WaterSensor first, WaterSensor second) throws Exception {
                   return first + "===" + second;
               }
           })
           .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
    
}
